package com.wudl.windows;


import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;


/**
 * @ClassName : Window_TimeWindow
 * @Description : 时间窗口
 * @Author :wudl
 * @Date: 2020-10-24 01:26
 */

public class Window_TimeWindow {
    public static void main(String[] args) throws Exception {

        // 1. 获取上下文的环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.设置并行度
        env.setParallelism(1);
        DataStream<String> dataStream = env.socketTextStream("192.168.1.180", 8899);

        KeyedStream<Tuple2<String, Integer>, String> dataKS = dataStream.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
                return Tuple2.of(s, 1);
            }
        }).keyBy(k -> k.f0);

//        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowedStream = dataKS.timeWindow(Time.seconds(5));
        // 滚动窗口 => 传一个参数： 窗口长度
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowedStream = dataKS.timeWindow(Time.seconds(5),Time.seconds(2));
//        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowedStream = dataKS.timeWindow(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(2));
//        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowedStream = dataKS.timeWindow(ProcessingTimeSessionWindows.withGap(Time.seconds(3)));



        windowedStream.sum(1).print();
        env.execute();

    }
}
